package com.we.risk.behavior.cross;

import com.alibaba.fastjson.JSONObject;
import com.we.utils.CommonTimeTools;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.we.risk.behavior.cross.CrossCommonVars.DATERANGE;

public class TableFdmHfqLoanCenterRiskRecordHive {
    public static final String KFKTABLE = "cdh6_hive.fdm.kafka_rtd_hfq_loancenter_risk_record_by230425";

    public static final String QUERYSQL = "select user_key, loan_id, " +
            "loan_line, loan_avail_line, loan_rate, " +
            "apply_period, apply_amount, " +
            "borrow_time as apply_time, " +
            "loan_period, uid " +
            "from " + KFKTABLE;

    public String user_key;

    public String loan_id;

    @DataTypeHint("DECIMAL(19, 6)")
    public BigDecimal loan_line;

    @DataTypeHint("DECIMAL(19, 6)")
    public BigDecimal loan_avail_line;

    @DataTypeHint("DECIMAL(19, 6)")
    public BigDecimal loan_rate;

    public Integer apply_period;

    @DataTypeHint("DECIMAL(19, 6)")
    public BigDecimal apply_amount;

    public Long apply_time;

    public Integer loan_period;

    public Long uid;

    public static Logger LOG = LoggerFactory.getLogger(TableFdmHfqLoanCenterRiskRecordHive.class);

    public TableFdmHfqLoanCenterRiskRecordHive() {}

    public TableFdmHfqLoanCenterRiskRecordHive(String user_key, String loan_id, BigDecimal loan_line,
                                               BigDecimal loan_avail_line, BigDecimal loan_rate,
                                               Integer apply_period, BigDecimal apply_amount,
                                               Long apply_time, Integer loan_period, Long uid) {
        this.user_key = user_key;
        this.loan_id = loan_id;
        this.loan_line = loan_line;
        this.loan_avail_line = loan_avail_line;
        this.loan_rate = loan_rate;
        this.apply_period = apply_period;
        this.apply_amount = apply_amount;
        this.apply_time = apply_time;
        this.loan_period = loan_period;
        this.uid = uid;
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(QUERYSQL);
        DataStream<TableFdmHfqLoanCenterRiskRecordHive> inputStream =
                tableEnv.toDataStream(tableRes, TableFdmHfqLoanCenterRiskRecordHive.class);
        SingleOutputStreamOperator<String> out = inputStream
        .flatMap(new RichFlatMapFunction<TableFdmHfqLoanCenterRiskRecordHive, String>() {
            SimpleDateFormat sdf = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                this.sdf = new SimpleDateFormat("yyyy-MM-dd");
            }

            @Override
            public void flatMap(TableFdmHfqLoanCenterRiskRecordHive value, Collector<String> out) throws Exception {
                long apply_time = value.getApply_time().longValue() * 1000L;
                long currentTime = System.currentTimeMillis();
                Integer timediff = CommonTimeTools.dateDiff(currentTime, apply_time, sdf);
                if (timediff.intValue() <= DATERANGE) {
                    out.collect(consResultJson(value));
                }
            }

            private String consResultJson(TableFdmHfqLoanCenterRiskRecordHive record) {
                JSONObject resJson = new JSONObject();
                resJson.put("user_key", record.getUser_key());
                resJson.put("loan_id", record.getLoan_id());
                resJson.put("loan_line", record.getLoan_line());
                resJson.put("loan_avail_line", record.getLoan_avail_line());
                resJson.put("loan_rate", record.getLoan_rate());
                resJson.put("apply_period", record.getApply_period());
                resJson.put("apply_amount", record.getApply_amount());
                resJson.put("apply_time", record.getApply_time());
                resJson.put("loan_period", record.getLoan_period());
                resJson.put("uid", record.getUid());
                return resJson.toString();
            }
        });
        return out;
    }

    public Long getUid() {
        return this.uid;
    }

    public String getUser_key() {
        return this.user_key;
    }

    public String getLoan_id() {
        return this.loan_id;
    }

    public BigDecimal getLoan_line() {
        return this.loan_line;
    }

    public Long getApply_time() {
        return this.apply_time;
    }

    public BigDecimal getLoan_avail_line() {
        return this.loan_avail_line;
    }

    public BigDecimal getLoan_rate() {
        return this.loan_rate;
    }

    public Integer getApply_period() {
        return this.apply_period;
    }

    public BigDecimal getApply_amount() {
        return this.apply_amount;
    }

    public Integer getLoan_period() {
        return this.loan_period;
    }
}
